package kafkaU

import (
	"crypto/tls"
	"crypto/x509"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/plain"
	"io/ioutil"
	"log"
	"time"
)

var (
	brokers  = []string{"192.168.20.31:9092"}
	product  *kafka.Writer
	consumer *kafka.Reader
	dialer   *kafka.Dialer
)

// to produce messages
func initKafkaConfig() {
	mechanism := plain.Mechanism{
		Username: "producer",
		Password: "prod-sec@Gd~CTrH]-sV[g]h",
	}
	tlsConfig, err := newTLSConfigP("kafkaU/certificate.pem", "kafkaU/key.pem", "kafkaU/CARoot.pem")
	if err != nil {
		log.Fatal("Certificate generation failed", err)
	}
	// This can be used on test server if domain does not match cert:
	tlsConfig.InsecureSkipVerify = true
	dialer = &kafka.Dialer{
		Timeout:       10 * time.Second,
		DualStack:     true,
		TLS:           tlsConfig,
		SASLMechanism: mechanism,
	}
}

func initKafkaProduct() *kafka.Writer {
	if dialer == nil {
		initKafkaConfig()
	}
	w := kafka.NewWriter(kafka.WriterConfig{
		Brokers: brokers,
		//Topic:    "topic-A",
		Balancer: &kafka.Hash{},
		Dialer:   dialer,
	})
	return w
}

func initKafkaConsumer(topic, groupId string) *kafka.Reader {
	if dialer == nil {
		initKafkaConfig()
	}
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:        brokers,
		GroupID:        groupId,
		Topic:          topic,
		Dialer:         dialer,
		CommitInterval: 10 * time.Second,
	})
	return r
}

// NewTLSConfigP NewTLSConfig generates a TLS configuration used to authenticate on server with
// certificates.
// Parameters are the three pem files path we need to authenticate: client cert, client key and CA cert.
func newTLSConfigP(clientCertFile, clientKeyFile, caCertFile string) (*tls.Config, error) {
	tlsConfig := tls.Config{}

	// Load client cert
	cert, err := tls.LoadX509KeyPair(clientCertFile, clientKeyFile)
	if err != nil {
		return &tlsConfig, err
	}
	tlsConfig.Certificates = []tls.Certificate{cert}

	// Load CA cert
	caCert, err := ioutil.ReadFile(caCertFile)
	if err != nil {
		return &tlsConfig, err
	}
	caCertPool := x509.NewCertPool()
	caCertPool.AppendCertsFromPEM(caCert)
	tlsConfig.RootCAs = caCertPool
	tlsConfig.BuildNameToCertificate()
	return &tlsConfig, err
}

// GetKafkaProduct 获取Kafka生产者
func GetKafkaProduct() *kafka.Writer {
	if product == nil {
		return initKafkaProduct()
	}
	return product
}

// GetKafkaConsumer 获取Kafka消费者
func GetKafkaConsumer(topic, groupId string) *kafka.Reader {
	if consumer == nil {
		return initKafkaConsumer(topic, groupId)
	}
	return consumer
}
